package com.mnb.beidou.config;

import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.support.ErrorMessage;

import com.mnb.beidou.service.MessageService;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Configuration
public class MqttConfig {
    
    private static final Logger logger = LoggerFactory.getLogger(MqttConfig.class);
    
    @Value("${mqtt.broker.url}")
    private String brokerUrl;
    
    @Value("${mqtt.broker.port}")
    private int brokerPort;
    
    @Value("${mqtt.client.id}")
    private String clientId;
    
    @Value("${mqtt.username}")
    private String username;
    
    @Value("${mqtt.password}")
    private String password;
    
    @Value("${mqtt.topic}")
    private String topic;
    
    @Value("${mqtt.protocol.version:3.1.1}")
    private String protocolVersion;
    
    @Bean
    @ConditionalOnProperty(name = "mqtt.enabled", havingValue = "true", matchIfMissing = true)
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        
        try {
            // 设置服务器地址和端口 - 支持多种协议
            String serverUri;
            if (brokerUrl.startsWith("ssl://") || brokerUrl.startsWith("tcp://") || brokerUrl.startsWith("mqtt://")) {
                // 如果已经包含协议前缀，直接使用
                serverUri = brokerUrl + ":" + brokerPort;
            } else {
                // 默认使用tcp协议，包括mgtt.bds100.com域名
                serverUri = "tcp://" + brokerUrl + ":" + brokerPort;
            }
            logger.info("尝试连接MQTT服务器: {}", serverUri);
            options.setServerURIs(new String[] { serverUri });
            
            // 设置认证信息
            if (username != null && !username.trim().isEmpty()) {
                options.setUserName(username);
                options.setPassword(password.toCharArray());
            }
            
            // 使用保守的自动重连设置
            options.setConnectionTimeout(60);   // 标准连接超时
            options.setKeepAliveInterval(60);   // 标准心跳间隔
            options.setAutomaticReconnect(true); // 启用自动重连
            options.setCleanSession(true);      // 清理会话
            options.setMaxInflight(1000);       // 标准并发数
            options.setMaxReconnectDelay(30000); // 最大重连延迟30秒
            
            // 设置MQTT协议版本 - 根据配置设置
            if ("3.1.1".equals(protocolVersion)) {
                options.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1_1);
                logger.info("MQTT协议版本设置为: 3.1.1");
            } else if ("3.1".equals(protocolVersion)) {
                options.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1);
                logger.info("MQTT协议版本设置为: 3.1");
            } else {
                // 默认使用3.1.1版本
                options.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1_1);
                logger.info("MQTT协议版本设置为默认: 3.1.1");
            }
            
            // 设置遗嘱消息（可选）
            // options.setWill("client/status", "offline".getBytes(), 1, false);
            
            // 添加连接状态监听器
            factory.setConnectionOptions(options);
            
            logger.info("MQTT客户端工厂配置完成，服务器地址: {}:{}", brokerUrl, brokerPort);
            logger.info("MQTT连接参数 - 协议版本: {}, 超时: {}秒, 心跳: {}秒, 自动重连: {}, 清理会话: {}", 
                       protocolVersion, options.getConnectionTimeout(), 
                       options.getKeepAliveInterval(), options.isAutomaticReconnect(), options.isCleanSession());
            return factory;
        } catch (Exception e) {
            logger.error("MQTT客户端工厂配置失败: {}", e.getMessage(), e);
            throw e;
        }
    }
    
    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }
    
    @Bean
    public MessageChannel mqttErrorChannel() {
        return new DirectChannel();
    }
    
    @Bean
    @ConditionalOnProperty(name = "mqtt.enabled", havingValue = "true", matchIfMissing = true)
    public MqttPahoMessageDrivenChannelAdapter mqttAdapter() {
        try {
            // 暂时不订阅任何主题，只测试连接
            String subscribeTopic = (topic == null || topic.trim().isEmpty()) ? "" : topic;
            
            // 处理主题订阅
            String[] topics;
            if (subscribeTopic == null || subscribeTopic.trim().isEmpty()) {
                // 不订阅任何主题，只建立连接
                topics = new String[0];
                logger.info("不订阅任何主题，仅建立MQTT连接");
            } else {
                // 支持多个主题订阅，用逗号分隔
                topics = subscribeTopic.split(",");
                for (int i = 0; i < topics.length; i++) {
                    topics[i] = topics[i].trim();
                }
            }
            
            // 使用简单的客户端ID，避免时间戳冲突
            String uniqueClientId = clientId + "_spring_" + (System.currentTimeMillis() % 10000);
            
            MqttPahoMessageDrivenChannelAdapter adapter = 
                new MqttPahoMessageDrivenChannelAdapter(uniqueClientId, mqttClientFactory(), topics);
            
            // 使用保守的适配器重连设置
            adapter.setCompletionTimeout(15000);  // 15秒完成超时
            adapter.setConverter(new DefaultPahoMessageConverter());
            adapter.setQos(1);                    // QoS 1，确保消息可靠性
            adapter.setOutputChannel(mqttInputChannel());
            
            // 设置保守的重连间隔
            adapter.setRecoveryInterval(30000);   // 30秒重连间隔
            adapter.setErrorChannelName("mqttErrorChannel");
            
            // 添加连接状态日志
            logger.info("MQTT适配器重连配置 - 完成超时: 15秒, 重连间隔: 30秒, QoS: 1");
            
            logger.info("MQTT消息适配器配置完成，客户端ID: {}, 订阅主题: {}", uniqueClientId, String.join(", ", topics));
            return adapter;
        } catch (Exception e) {
            logger.error("MQTT适配器配置失败: {}", e.getMessage(), e);
            throw e;
        }
    }
    
    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler(MessageService beidouMessageService) {
        return message -> {
            try {
                Object payload = message.getPayload();
                String messageContent;
                
                // 处理不同类型的消息载荷
                if (payload instanceof byte[]) {
                    messageContent = new String((byte[]) payload, "UTF-8");
                } else if (payload instanceof String) {
                    messageContent = (String) payload;
                } else {
                    messageContent = payload.toString();
                }
                
                // 获取主题信息
                String topic = message.getHeaders().get("mqtt_receivedTopic", String.class);
                
                // 检查是否为Retain消息
                boolean isRetain = message.getHeaders().containsKey("mqtt_receivedRetained") &&
                                 (Boolean) message.getHeaders().get("mqtt_receivedRetained");
                
                if (isRetain) {
                    logger.info("收到Retain消息，跳过处理 - 主题: {}, 内容: {}", topic, messageContent);
                    return; // 跳过Retain消息的处理
                }
                
                // 调用消息处理服务，传递主题信息
                beidouMessageService.processMessage(messageContent, topic);
                
            } catch (Exception e) {
                logger.error("处理MQTT消息时发生错误", e);
            }
        };
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttErrorChannel")
    public MessageHandler errorHandler() {
        return message -> {
            if (message instanceof ErrorMessage) {
                ErrorMessage errorMessage = (ErrorMessage) message;
                Throwable throwable = errorMessage.getPayload();
                logger.error("MQTT连接错误: {}", throwable.getMessage(), throwable);
                
                // 可以在这里添加错误处理逻辑，比如发送告警通知
                if (throwable instanceof org.eclipse.paho.client.mqttv3.MqttException) {
                    org.eclipse.paho.client.mqttv3.MqttException mqttException = 
                        (org.eclipse.paho.client.mqttv3.MqttException) throwable;
                    logger.error("MQTT异常代码: {}", mqttException.getReasonCode());
                }
            }
        };
    }
} 